package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.app.func.BeanToJsonStrFunction;
import realtime.app.func.DimAsyncFunction;
import realtime.beans.TradeProvinceOrderBean;
import realtime.util.DateFormatUtil;
import realtime.util.DorisUtil;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

import java.math.BigDecimal;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
/*
 * string->jsonObject
 * 水位线
 * 来源分组 用户状态 订单状态 列表状态，状态中只有一个对象，对象只有一个属性set，jsonobject添加用户、订单数量
 * 开窗
 * 实体类对象
 * 聚合
 * */


public class DwsTradeProvinceWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_province_group";
        KafkaSource<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaSourceDS = env.fromSource(kafkaConsumer, WatermarkStrategy.noWatermarks(), "kafka_source");
        //转换数据类型 jsonobject
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaSourceDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(value)) {
                            JSONObject jsonObject = JSON.parseObject(value);
                            out.collect(jsonObject);
                        }
                    }
                }
        );
        //指定水位线
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = jsonDS.assignTimestampsAndWatermarks(WatermarkStrategy.
                <JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts")*1000;
                    }
                }));
        //省份id分组
        KeyedStream<JSONObject, String> keyedByDS = jsonDS.keyBy(new KeySelector<JSONObject, String>() {

            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("province_id");
            }
        });

        //keyedByDS.print();
        //转换为实体类对象 用户状态 订单状态 值状态，状态中只有一个hashset
        SingleOutputStreamOperator<TradeProvinceOrderBean> mapDS = keyedByDS.map(new RichMapFunction<JSONObject, TradeProvinceOrderBean>() {
            private ValueState<HashSet> userState;
            private ValueState<HashSet> orderState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<HashSet> userStateDescriptor = new ValueStateDescriptor<HashSet>("user", HashSet.class);
                userStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                userState = getRuntimeContext().getState(userStateDescriptor);

                ValueStateDescriptor<HashSet> orderStateDescriptor = new ValueStateDescriptor<HashSet>("order", HashSet.class);
                orderStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                orderState = getRuntimeContext().getState(orderStateDescriptor);
            }

            @Override
            public TradeProvinceOrderBean map(JSONObject value) throws Exception {
                String ud = value.getString("user_id");
                String od = value.getString("order_id");
                if (userState.value() != null) {
                    HashSet<String> u = userState.value();
                    u.add(ud);
                    userState.update(u);
                } else {
                    HashSet<String> u = new HashSet<>();
                    u.add(ud);
                    userState.update(u);
                }

                if (orderState.value() != null) {
                    HashSet<String> o = orderState.value();
                    o.add(od);
                    orderState.update(o);
                } else {
                    HashSet<String> o = new HashSet<>();
                    o.add(od);
                    orderState.update(o);
                }

                String provinceId = value.getString("province_id");
                String originAmount = value.getString("origin_amount");
                Long uc = 0l;
                uc = (long)userState.value().size();
                Long oc = 0l;
                oc = (long)orderState.value().size();
                Long ts = value.getLong("ts");
                TradeProvinceOrderBean tradeProvinceOrderBean = TradeProvinceOrderBean.builder()
                        .provinceId(provinceId)
                        .orderAmount(new BigDecimal(originAmount))
                        .orderUserCt(uc)
                        .orderCt(oc)
                        .ts(ts)
                        .build();
                return tradeProvinceOrderBean;
            }
        });
        //mapDS.print();

        //省份id分组 与 window 开窗，没有分组只能windowall，只有一个并行度
        KeyedStream<TradeProvinceOrderBean, String> keyBy = mapDS.keyBy(new KeySelector<TradeProvinceOrderBean, String>() {
            @Override
            public String getKey(TradeProvinceOrderBean value) throws Exception {
                return value.getProvinceId();
            }
        });
//        keyBy.print();
        WindowedStream<TradeProvinceOrderBean, String, TimeWindow> windowDS = keyBy.window(TumblingProcessingTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(1)));
        //聚合
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setOrderCt(value1.getOrderCt().longValue()>value2.getOrderCt().longValue()?value1.getOrderCt():value2.getOrderCt());
                        value1.setOrderUserCt(value1.getOrderUserCt().longValue()>value2.getOrderUserCt().longValue()?value1.getOrderUserCt():value2.getOrderUserCt());
//                        System.out.println("reduce");
//                        System.out.println(value1);
                        return value1;
                    }
                },
                new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderBean> input, Collector<TradeProvinceOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        long timeMillis = System.currentTimeMillis();
                        String curDate = DateFormatUtil.toDate(timeMillis);

                        for (TradeProvinceOrderBean tradeProvinceOrderBean : input) {
                            tradeProvinceOrderBean.setStt(stt);
                            tradeProvinceOrderBean.setEdt(edt);
                            tradeProvinceOrderBean.setCurDate(curDate);
                            out.collect(tradeProvinceOrderBean);
                        }
                    }
                });
        //reduceDS.print();
        //关联省份维度
        SingleOutputStreamOperator<TradeProvinceOrderBean> provinceDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeProvinceOrderBean>("dim_base_province") {
                    @Override
                    public String getKey(TradeProvinceOrderBean bean) {
                        return bean.getProvinceId();
                    }

                    @Override
                    public void join(TradeProvinceOrderBean bean, JSONObject dim) {
                        bean.setProvinceName(dim.getString("name"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //provinceDS.print();
        provinceDS.map(new BeanToJsonStrFunction<>())
                .sinkTo(DorisUtil.getDorisSink("dws_trade_province_window"));
        env.execute();
    }
}
